
package com.lee.lossycounting; 
 
import com.google.common.collect.Lists; 
import com.lee.enumeration.FinishQualifierEnum; 
import com.lee.info.CkFlowInfo; 
import com.lee.info.LkFlowInfo; 
import com.lee.logger.Logger; 
import com.lee.utils.*; 
import org.apache.flink.api.common.functions.FlatMapFunction; 
import org.apache.flink.util.Collector; 
 
import java.time.Duration; 
import java.time.LocalDateTime; 
import java.util.ArrayList; 
import java.util.List; 
import java.util.Objects; 
 
public class ScanDFlowFunction implements FlatMapFunction<CkFlowInfo, LkFlowInfo> { 
 
 
 
    @Override
    public void flatMap(CkFlowInfo ckFlowInfo, Collector<LkFlowInfo> collector) {
        try {
 
            Logger.debug(ckFlowInfo.getDIndex() ,"/",ckFlowInfo.getDCIndex(),"LossyCounting ScanDFlowFunction -> ckFlowInfo ", ckFlowInfo);
            LocalDateTime start = LocalDateTime.now();
            List<String> Lk = new ArrayList<>();
            long index = ckFlowInfo.getDIndex();
            long cindex = ckFlowInfo.getDCIndex();
            long num = DCountUtils.getIndex();
            List<String> CkFlowItem = ckFlowInfo.getCkList();
            if(Objects.isNull(CkFlowItem)){
                return;
            }
 
            //lossycounting的频繁度计算时需要考虑块数
            long blockNum = index / ConstantsUtils.DATA_BLOCK_SIZE;
            //1、从汇总表中查询支持度计数
            long count = SupportDataUtils.getValueByKeyNum(CkFlowItem);
            count += 1;
            //2、根据查询出来的次数更新支持度
            double support = 1.0 * (count + blockNum) / num;

 
            SupportDataUtils.setValueByKey(CkFlowItem, support);
//            SupportDataUtils.setValueByKeyNum(CkFlowItem, count);
            Logger.info(index, " changed LossyCounting BdsScanDFlowFunction -> CkFlowItem ", CkFlowItem, " len ", count, " num ", num, " support ", support);
            Lk = CkFlowItem;
 
 
            LkFlowInfo lkFlowInfo = new LkFlowInfo();
            lkFlowInfo.setDIndex(index);
            lkFlowInfo.setDCIndex(cindex);
            lkFlowInfo.setLkList(Lk);
            lkFlowInfo.setDataFlow(ckFlowInfo.getDataFlow());
            collector.collect(lkFlowInfo);
 
            DMapUtils.addFinishIndex(FinishQualifierEnum.SCAN_D.getCode(),1l);
            LocalDateTime end = LocalDateTime.now();
            Duration between = Duration.between(start, end);
            Logger.debug(ckFlowInfo.getDIndex() ,"/",ckFlowInfo.getDCIndex(),between.toMillis(),"END LossyCounting ScanDFlowFunction -> ckFlowInfo ", ckFlowInfo);
 
        }catch(Exception e){
            Logger.error("LossyCounting BdsScanDFlowFunction -> 支持度计算报错：",e);
        }
 
    }
} 
